package org.example.service.impl;

import cn.hutool.core.util.IdUtil;
import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.vertx.core.Vertx;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.net.NetClient;
import io.vertx.core.net.NetClientOptions;
import io.vertx.core.net.NetSocket;
import io.vertx.mqtt.MqttClient;
import lombok.extern.slf4j.Slf4j;
import org.example.domain.Sell;
import org.example.domain.Stock;
import org.example.domain.TakePosition;
import org.example.domain.User;
import org.example.mapper.TakePositionMapper;
import org.example.service.SellService;
import org.example.service.StockService;
import org.example.service.TakePositionService;
import org.example.service.UserService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.io.IOException;
import java.math.BigDecimal;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Date;
import java.util.List;

/**
 * @author scf
 * @description 针对表【take_position】的数据库操作Service实现
 * @createDate 2024-06-21 09:53:44
 */
@Slf4j
@Service
public class TakePositionServiceImpl extends ServiceImpl<TakePositionMapper, TakePosition>
        implements TakePositionService {
    @Autowired
    private StockService stockService;
    @Autowired
    private UserService userService;
    @Autowired
    private SellService sellService;
    private String ip = "127.0.0.1";
    private int port = 1883;
    private MqttClient mqttClient;

    @Override
    @Transactional
    public String buy(TakePosition takePosition, Integer userId) {
        Stock stock = stockService.getById(takePosition.getStockId());
        TakePosition takePosition1 = new TakePosition();
        takePosition1.setStockId(takePosition.getStockId());
        takePosition1.setUserId(userId);
        //最新价*购买数量
        takePosition1.setVolumePrice(stock.getLatestPrice().multiply(new BigDecimal(takePosition.getNum())));
        takePosition1.setNum(takePosition.getNum());
        takePosition1.setCreateDate(new Date());
        takePosition1.setVolumeCode(IdUtil.getSnowflake().nextIdStr());
        takePosition1.setAveragePrice(stock.getLatestPrice());
        //如果数量为0则不能购买
        if (takePosition.getNum() == 0) {
            return "还未进行购买";
        }
        //用户余额不足
        User user = userService.getById(userId);
        if (user.getMoney().compareTo(takePosition1.getVolumePrice()) < 0) {
            return "用户余额不足";
        }
        boolean save = this.save(takePosition1);
        UDPClient(takePosition1);
        MQTTClient(takePosition1);
        //减去用户余额
        user.setMoney(user.getMoney().subtract(takePosition1.getVolumePrice()));
        save = userService.updateById(user);
        if (stock.getTradingVolume() == null) {
            stock.setTradingVolume(1);
            stock.setClosingDate(new Date());
            stock.setVolumeBusiness(takePosition1.getVolumePrice());
            stockService.updateById(stock);
            return "购买成功";
        }
        stock.setTradingVolume(stock.getTradingVolume() + 1);
        stock.setVolumeBusiness(takePosition1.getVolumePrice());
        stock.setClosingDate(new Date());
        stockService.updateById(stock);
        return "购买成功";
    }

    @Override
    public List<TakePosition> listall(Integer userId, Integer stockId) {
        return baseMapper.listAll(userId, stockId);
    }

    @Override
    @Transactional
    public String buyorThree(TakePosition takePosition, Integer userId, String volumeCode) {
        User user = userService.getById(takePosition.getUserId());
        //成功买入  添加到持仓中
        boolean save = this.save(takePosition);
        if (save) {
            //减去用户余额
            user.setMoney(user.getMoney().subtract(takePosition.getVolumePrice()));
            userService.updateById(user);
        }
        //委托人资金增加
        User one = userService.getById(userId);
        if (one != null) {
            one.setMoney(one.getMoney().add(takePosition.getVolumePrice()));
            userService.updateById(one);
        }
        //委托人委托改为已成交
        Sell sell2 = sellService.getsellByuuid(volumeCode);
        if (sell2 != null) {
            sell2.setIssell(2);
            sellService.updateById(sell2);
        }
        return "购买成功";
    }

    @Override
    public TakePosition getOneByVolumeCode(String volumeCode) {
        return baseMapper.getOneByVolumeCode(volumeCode);
    }

    @Override
    public void TCPClient(Sell sell) {
        Vertx vertx = Vertx.vertx();
        NetClientOptions options = new NetClientOptions()
                .setConnectTimeout(10000)
                .setReconnectAttempts(10)
                .setReconnectInterval(500)
                .setLogActivity(true);
        NetClient client = vertx.createNetClient(options);
        client.connect(8891, "127.0.0.1", res -> {
            if (res.succeeded()) {
                System.out.println("连接成功");
                NetSocket socket = res.result();
                try {
                    String s = JSONUtil.toJsonStr(sell);
                    int length = s.length();
                    System.out.println(length);
                    Buffer buffer = Buffer.buffer().appendInt(length).appendString(s);
                    socket.write(buffer);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            } else {
                System.out.println("连接失败");
            }
        });

    }

    public void UDPClient(TakePosition takePosition) {
        try (DatagramSocket socket = new DatagramSocket()) {
            String message = "BUY:" + takePosition.getStockId() + ":" + takePosition.getNum() + ":" + takePosition.getAveragePrice();
            byte[] data = message.getBytes();
            InetAddress addresses = InetAddress.getByName("127.0.0.1");
            int port = 8085;
            DatagramPacket datagramPacket = new DatagramPacket(data, data.length, addresses, port);
            socket.send(datagramPacket);
            System.out.println("发送成功");
        } catch (UnknownHostException e) {
            throw new RuntimeException(e);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    public void MQTTClient(TakePosition takePosition) {
        Vertx vertx = Vertx.vertx();
        mqttClient = MqttClient.create(vertx);
        mqttClient.connect(port, ip, res -> {
            if (res.succeeded()) {
                log.info("connect to mqtt ip:{},port:{}", "127.0.0.1", 1883);
                String topic = "MyTopic";
                String s = JSONUtil.toJsonStr(takePosition);
                mqttClient.publish(topic, Buffer.buffer(s), MqttQoS.AT_MOST_ONCE, false, false);

            } else {
                log.warn("[fail] connect to mqtt ip:{},port:{}", ip, port);
            }
        }).exceptionHandler(throwable -> System.out.println("连接mqtt异常"));
    }

}




